package com.tod.spark.springbootspark.sql;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

import java.util.ArrayList;
import java.util.List;
import java.util.regex.Pattern;

public class RDD2DataFrameDynamic {
    private static final Pattern SPACE = Pattern.compile(",");

    public static void main(String[] args) {

        SparkConf conf = new SparkConf().setAppName("dataframe").setMaster("local");
        JavaSparkContext jsc = new JavaSparkContext(conf);
        SQLContext sqlContext = new SQLContext(jsc);

        JavaRDD<String> lines = jsc.textFile("src/main/resources/static/student.txt");
        JavaRDD<Row> rows  = lines.map(line -> {
            String[] lineSplit = SPACE.split(line);
            return RowFactory.create(
                    Integer.valueOf(lineSplit[0]),
                    lineSplit[1],
                    Integer.valueOf(lineSplit[2]));
        });

        //动态构造数据，我们没有JavaBean让你反射推断出来那些Field，但是我们可以通过这种方式fields类型可以从mysql或配置文件中加载出来
        List<StructField> fields = new ArrayList<>();
        fields.add(DataTypes.createStructField("id",DataTypes.IntegerType,true));
        fields.add(DataTypes.createStructField("name",DataTypes.StringType,true));
        fields.add(DataTypes.createStructField("age",DataTypes.IntegerType,true));

        StructType structType = DataTypes.createStructType(fields);
        Dataset<Row> studentDS = sqlContext.createDataFrame(rows, structType);
        studentDS.registerTempTable("students");
        Dataset<Row> teenagerDS = sqlContext.sql("select * from students where age <= 18");
        List<Row> students = teenagerDS.javaRDD().collect();
        System.out.println("***************************************************");
        students.forEach(row -> System.out.println(row));
        System.out.println("***************************************************");

        jsc.stop();
    }
}
